package com.code.task.service;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.TimeInterval;
import com.code.common.utils.collection.ListUtil;
import com.code.task.worker.ItemIndexWorker;
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import java.util.List;
import java.util.concurrent.*;
import java.util.stream.Collectors;

/**
 * @author lyf
 * @date 2021-06-15 下午3:39
 */
@Slf4j
@Service
public class IndexTaskService {

    // 异步线程future 容器   因为这行代码报错，所以使用了下一行的创建方式
    // private List<Future<Boolean>> keepFutures = Lists.newCopyOnWriteArrayList();
    private List<Future<Boolean>> keepFutures = Lists.newArrayList();

    // 自定义线程池
    private ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(5, 20, 2, TimeUnit.SECONDS, new LinkedBlockingQueue<>(5));

    @Autowired
    private ItemIndexService itemIndexService;

    /**
     * 根据 index 当前分片参数获取当前机器待执行的门店列表集合 storeIds
     * 分n个线程[线程数从远程配置中心读取 factor]跑数据
     *
     * @param total
     * @param index
     * @return
     */
    public boolean indexAll(Integer total, Integer index) {
        TimeInterval timer = DateUtil.timer();

        // 查询所有门店id
        List<Long> storeIds = Lists.newArrayList(1L, 2L, 3L);

        // 当前分片 需要索引的门店id列表
        List<Long> currentShardStoreIds = storeIds.stream()
                .filter(storeId -> Math.floorMod(storeId, total.longValue()) == index)
                .collect(Collectors.toList());

        try {
            // 获取本地执行线程数量
            // String factoryStr = environment.getProperty("indexAll.shard.factor");
            String factoryStr = "4";
            Integer factor = StringUtils.isEmpty(factoryStr) ? 4 : Integer.parseInt(factoryStr);

            if (CollectionUtils.isEmpty(keepFutures)) {
                // 平均分为factor份
                List<List<Long>> storeIdLists = ListUtil.averageAssign(currentShardStoreIds, factor);
                for (List<Long> ids : storeIdLists) {
                    keepFutures.add(poolExecutor.submit(new ItemIndexWorker(ids, itemIndexService)));
                }
                log.info("ItemIndexService.indexAll the factor is {}, keepFutures size is {}", factor, keepFutures.size());

                for (int i = 0; i < keepFutures.size(); i++) {
                    keepFutures.get(i).get();
                }
                //运行结束，清空成员变量
                keepFutures.clear();
                log.info("ItemIndexService.indexAll success, keepFutures size is {}", keepFutures.size());
            }
        } catch (InterruptedException e) {
            log.error("ItemIndexService.indexAll InterruptedException error", e);
            Thread.currentThread().interrupt();
        } catch (ExecutionException e) {
            log.error("ItemIndexService.indexAll ExecutionException error", e);
        } catch (Exception e) {
            log.error("ItemIndexService.indexAll Exception error", e);
        }
        log.info("ItemIndexService is finished ,the total is {},the index is {} ,cost time is{}", total, index, timer.interval());
        return true;
    }

    public boolean cancelIndexAll() {
        if (!CollectionUtils.isEmpty(keepFutures)) {
            for (Future<Boolean> future : keepFutures) {
                if (!future.isCancelled()) {
                    try {
                        future.cancel(true);
                    } catch (Exception e) {
                        log.error("cancelIndexAll error", e);
                    }

                }
            }
            keepFutures.clear();
        }
        return true;
    }

}
